package com.bigdata.sql

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
import org.apache.spark.storage.StorageLevel

/**
 * 从HDFS读写数据，但是我们只指定表名，不指定HDFS文件路径
 * 从过metastore服务（桥梁）连接，可以知道表名和字段对应的是HDFS中的哪些数据
 */
object Demo10SparkOnHive {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .appName("onhive")
      .master("local")
      // 设置连接hive的metastore服务，打通元数据
      // spark on hive(metastore)
      //.config("hive.metastore.uris", "thrift://node03:9083")
      //.config("hive.exec.dynamic.partition.mode", "nonstrict")
      //开启hive的元数据支持，在代码中读取hive的元数据
      .enableHiveSupport()
      .getOrCreate()


    //    val studentDF: DataFrame = spark
    //      .read
    //      .format("json")
    //      .load("data/students.json")

    // 创建一个临时表
    // studentDF.createOrReplaceTempView("t_stu")

    import org.apache.spark.sql.functions._
    import spark.implicits._

    /**
     * 当spark sql和hive整合好之后可以在代码中直接读取hive中额表
     *
     */
    val df1: DataFrame = spark.sql(
      """
        |select * from bigdata.student where age <= 22
        |""".stripMargin)

    df1.show()

    // 也可以写出
    //    df1.printSchema()
    //    df1.show(false)

    // 通过spark-sql对hive表做的相关操作，会同步到metastore中
    // spark-sql wirte的表，在hive中不存在，会创建这张表，并且会在metastore中写入这张表的元数据信息
    df1
      .write
      .format("hive")
      .mode(SaveMode.Overwrite)
      .partitionBy("age")
    //      .saveAsTable("bigdata.stu_result_2")

    /**
     * 通过表名得到一个DF
     */
    val empDF: DataFrame = spark.table("bigdata.emp")

    import spark.implicits._
    import org.apache.spark.sql.functions._

    /**
     * 缓存DF
     *
     */
    // empDF.persist(StorageLevel.MEMORY_ONLY)

    empDF
      .groupBy($"deptno")
      .agg(count($"ename") as "count")
      // .show()

    Thread.sleep(10000000)


  }

}
